package cn.zhiquxing.lock.master_worker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class Master<T extends Task, R> {
    //所有worker的集合
    private Map<String, Worker<T, R>> workers = new HashMap<String, Worker<T, R>>();
    
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    
    // 任务处理结果合集
    protected Map<String, R> resultMap = new HashMap<>();
    
    //master的任务调度线程
    private Thread thread = null;
    
    private AtomicInteger sum = new AtomicInteger(0);
    
    public Master(int workerCount) {
        // 每个worker对象都需要持有queue的引用，用于领取任务与提交结果
        for (int i = 0; i < workerCount; i++) {
            Worker<T, R> worker = new Worker<>();
            workers.put("子节点" + i, worker);
        }
        thread = new Thread(() -> this.execute());
        thread.start();
    }
    
    public void submit(T task) {
        taskQueue.add(task);
    }
    
    public void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker:" + task.getWorkerId() + "-" + "Task:" + task.getId();
        R result = task.getResult();
        resultMap.put(taskName, result);
        sum.getAndAdd((Integer)result);
    }

    /**
     * 启动所有的子任务
     */
    public void execute() {
        for (;;) {
            for (Map.Entry<String, Worker<T, R>> entry : workers.entrySet()) {
                T task = null;
                try {
                    task = taskQueue.take();
                    Worker<T, R> worker = entry.getValue();
                    worker.submit(task, this::resultCallBack);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    public void printResult() {
        System.out.println("-----------------sum is:" + sum.get());
        for (Map.Entry<String, R> entry : resultMap.entrySet()) {
            System.out.println(entry.getKey() + ":" + entry.getValue());
        }
    }
}
